package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.CartAddUuBean;
import com.zhu.config.KafkaTopicConfig;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.time.Duration;

/**
 * DWS层独立用户加购聚合表
 * 过滤每天的独立用户加购  维护在状态中
 */
public class DWSTradeCartAddUuWindowApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);  //topic partition 4
        /*
        checkPoint
          streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次
        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L)
         */

        //todo kafka dwd_trade_cart_add

        String topic = KafkaTopicConfig.KAFKA_DWD_USER_CAST_ADD_TOPIC;
        String groupId = "dwd_trade_cart_add"+KafkaTopicConfig.KAFKA_GROUP_ID_LAST_NAME;

        DataStreamSource<String> kafkaDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo to json
        SingleOutputStreamOperator<JSONObject> jsonObjDStream = kafkaDStream.map(JSON::parseObject);

        //todo waterMark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWaterMarkDStream = jsonObjDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {   //可能会涉及到更新
                       // return DataFormatUtil.toTs(jsonObject.getString("create_time"), true);
                        String operateTime = jsonObject.getString("operate_time");
                        if(operateTime != null){
                            return DataFormatUtil.toTs(operateTime,true);
                        }else {
                            String createTime = jsonObject.getString("create_time");
                            return DataFormatUtil.toTs(createTime,true);
                        }
                    }
                }));

        //todo 状态 过滤
        //筛选每个独立用户状态

        KeyedStream<JSONObject, String> keyByUserIdDStream = jsonObjWithWaterMarkDStream.keyBy(jsonObject -> jsonObject.getString("user_id"));

        SingleOutputStreamOperator<CartAddUuBean> filterCartAddDStream = keyByUserIdDStream.flatMap(new RichFlatMapFunction<JSONObject, CartAddUuBean>() {

            private ValueState<String> lastCartAddDt;


            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ValueStateDescriptor<String> lastCartAddDescriptor =
                        new ValueStateDescriptor<String>("user_last_cart_add", String.class);
                lastCartAddDt = getRuntimeContext().getState(lastCartAddDescriptor);
            }

            //过滤出每个独立用户的状态
            @Override
            public void flatMap(JSONObject data, Collector<CartAddUuBean> collector) throws Exception {
                String operatorTime = data.getString("operator_time");
                String lastCreatTime = lastCartAddDt.value();
                String currDt = null;

                if(operatorTime != null){
                    currDt = operatorTime.split(" ")[0];
                }else {
                    currDt = data.getString("create_time").split(" ")[0];
                }

                if (lastCreatTime == null || !lastCreatTime.equals(currDt)){
                    lastCartAddDt.update(currDt);
                    collector.collect(new CartAddUuBean("","",1L,null));
                }
            }
        });

        //todo window
        AllWindowedStream<CartAddUuBean, TimeWindow> windowAllCartAddDStream = filterCartAddDStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));


        //todo reduce
        SingleOutputStreamOperator<CartAddUuBean> resultDStream = windowAllCartAddDStream.reduce(
                new ReduceFunction<CartAddUuBean>() {
                    @Override
                    public CartAddUuBean reduce(CartAddUuBean value1, CartAddUuBean value2) throws Exception {
                        value1.setCartAddCount(value1.getCartAddCount() + value2.getCartAddCount());
                        return value1;
                    }
                },
                new AllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<CartAddUuBean> iterable, Collector<CartAddUuBean> collector) throws Exception {
                        String start = DataFormatUtil.toYmdHms(timeWindow.getStart());
                        String end = DataFormatUtil.toYmdHms(timeWindow.getEnd());
                        long ts = System.currentTimeMillis();
                        CartAddUuBean bean = iterable.iterator().next();
                        bean.setTs(ts);
                        bean.setStt(start);
                        bean.setEdt(end);
                        collector.collect(bean);
                    }
                }
        );

        //todo clickHouse
        resultDStream.print(">>>>>");
       // resultDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction("" +
       //         "insert into  dws_trade_cart_add_uu_window values(?,?,?,?)"));

        //todo execute job
        streamExecutionEnvironment.execute();
    }
}
